package com.aotain.nyx.flink;

import java.util.Properties;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import com.aotain.nyx.util.CommonUtil;
import com.aotain.nyx.util.FilterNullEnum;
import com.aotain.nyx.util.FilterNullObject;

public class LocalFlinkTest {

	public static void main(String[] args) {
		// TODO Auto-generated method stub

		String dbJson = "{\"driverClassName\":\"com.mysql.jdbc.Driver\","
				+ "\"maxActive\":\"200\",\"maxIdle\":\"50\",\"maxWait\":\"10000\","
				+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
				+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:mysql://192.168.5.97:3306/SDS\""
				+ ",\"userName\":\"sds\""
				+ ",\"validationQuery\":\"SELECT SYSDATE()\"}";
		//String dbJson_oracle ="{\"driverClassName\":\"oracle.jdbc.driver.OracleDriver\","
		//		+ "\"maxActive\":\"200\",\"maxIdle\":\"50\",\"maxWait\":\"10000\","
		//		+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
		//		+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:oracle:thin:@192.168.50.155:1521:bsmp\","
		//		+ "\"userName\":\"sds\",\"validationQuery\":\"select sysdate from dual\"}";

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		Properties properties = new Properties();
		properties.setProperty("bootstrap.servers", "192.168.5.95:9092,192.168.5.96:9092,192.168.5.97:9092");
		// only required for Kafka 0.8
		properties.setProperty("zookeeper.connect", "192.168.5.97:2181,192.168.5.98:2181,192.168.5.95:2181/kafka_2");
		properties.setProperty("group.id", "flink");
		String filterIpSql = "SELECT DISTINCT A.IP FROM SDS_SERVERINFO A JOIN SDS_WEBSITEINFO B ON A.SITEID = B.SITEID WHERE B.MONITORSTATUS = 1";
		String whiteIpSql = " SELECT * FROM SDS_WHITELIST";

		DataStream<String> stream = env
				.addSource(new FlinkKafkaConsumer08<>("udii", new SimpleStringSchema(), properties))
				.map(new MapFunction<String, String>() {

					@Override
					public String map(String value) throws Exception {
						// TODO Auto-generated method stub
						return value;
					}

				});
		stream.print();
		//分流输出
		//		DataStream<Tuple2<String,String>> sourceStream = env.addSource(new FlinkKafkaConsumer08<>("udii", new SimpleStringSchema(), properties))
		//				.map(new SourceDataFilterMap(filterIpSql,whiteIpSql, dbJson))
		//				.filter(new FilterNullObject<String>(FilterNullEnum.F1ISNOTNULL));
		//
		//		SplitStream<Tuple2<String,String>> splitStream = sourceStream.split(new SourceDataSelector());
		//		DataStream<Tuple2<String, String>> dipStream = splitStream.select("DipStream")
		//				.map(new DipSoucreStreamStat())
		//				.setParallelism(8)
		//				.keyBy(0);
		//
		//		DataStream<Tuple2<String, String>> sipStream = splitStream.select("SipStream")
		//				.map(new DipSoucreStreamStat())
		//				.setParallelism(8)
		//				.keyBy(0);
		//		
		//		dipStream.print();
	}

}
